package com;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class CustomConsumer {

    @Test
    public void test1() {
        // 从指定的topic消费消息，消费者有所属的group
        String topic = "order";
        String groupId = "group_demo1";

        // 创建配置对象
        Properties props = new Properties();

        // 指定集群地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");


        // 消费者是按照offset偏移量来进行消费的，每条消息都有对应的offset，
        // offset按照从小到大消费，消费者会维护消费的offset，消费者默认按照上一次消费的offset继续消费
        // 自动提交模式值是true，提交的是offset
        // 消费者每次消费消息后，会把这条消息的offset自动提交kafka集群中
        // 会选择关闭自动提交，会带来重复消费的问题
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        // 设置消费者组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // 指定key、value的反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 创建kafkaconsumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 消费者订阅指定的topic
        consumer.subscribe(Arrays.asList(topic));


        // 持续性的拉取数据
        while (true) {
            // 消费者拉取消息, 在指定间隔时间后拉取一次，一次可能拉取多条数据，poll是以一个partition为单位来取数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMinutes(5));

            System.out.println("执行一次循环！！！！！");

            int i = 0;
            for (ConsumerRecord record : records) {
                // 1号分区里面有3条订单，在处理的时候，第二条订单发生了异常，会产生什么影响？
                if (record.partition() == 1) {
                    i++;
                    if (i == 2) {
                        System.out.println(1 / 0);
                        // 获取key，value， 。。。。。
                        System.out.println("topic:" + record.topic() + "---partition:" + record.partition()
                                + "---offset:" + record.offset() + "---key:" + record.key()
                                + "---value:" + record.value());
                    } else {
                        /*
                        如何保证幂等性：
                            结合业务逻辑来考虑：在每条消息中传递一个基于业务的唯一标识，比如订单消息，
                            一条订单消息理论上只需要消费一次，在消费的时候处理基于订单的业务，当这条订单相关的业务都正常处理完毕，
                            记录下来（db）；下次再消费订单的时候，有可能发生重复消费，判断一下这条订单是否已经处理完毕，如果处理过了，
                            直接过滤掉，处理后续的订单消息。
                         */
                        System.out.println("topic:" + record.topic() + "---partition:" + record.partition()
                                + "---offset:" + record.offset() + "---key:" + record.key()
                                + "---value:" + record.value());
                    }
                } else {
                    // 获取key，value， 。。。。。
                    System.out.println("topic:" + record.topic() + "---partition:" + record.partition()
                            + "---offset:" + record.offset() + "---key:" + record.key()
                            + "---value:" + record.value());
                }
                // 第一条消息就发生了异常，不会提交offset，下次会重复消费
                // 第二条消息发生了异常，会提交offset，下次消费的时候，发生异常的数据就无法再次消费了，消息丢失
                // consumer.commitSync();
            }
            // partition1,3条数据，第二条发生了异常，代码执行不到手动提交，再次启动consumer，会重复消费
            // 如果重复消费的问题解决了，能够达到消息恰好消费一次
            /*
            如果消息没有发生异常，并且consumer也不回宕机，那么还会有可能发生重复消费吗？
            在消费者消费消息的时候，需要在代码层面保证幂等性。
             */
            consumer.commitSync();
            System.out.println("手动提交！！！！");
        }

    }
}
